package com.mars.kafkademo.controller;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.mars.kafkademo.pojo.Message;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Optional;

/**
 * author by: mars
 * Date: 2020/4/7 14:10
 * Description: kafka消费者
 */
@Component
public class KafkaReceiver {

    @KafkaListener(topics = {"mars11"})
    public void listen(ConsumerRecord<?,?> record){
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            System.out.println("record:"+record);
            System.out.println("message:"+message);
        }
    }
}
